package com.shujia.stream

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object Demo3Window {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("wc")


    val sc: SparkContext = new SparkContext(conf)
    val ssc: StreamingContext = new StreamingContext(sc, Durations.seconds(5))

    //设置checkpoint
    ssc.checkpoint("spark/data/checkpoint")

    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 9999)

    val kvDS: DStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))

    /**
      * 统计最近15秒数量最多的单词，每隔5秒统计一次
      *
      * 滑动窗口中的时间窗口
      *
      */
    val windwsDS: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
      (x, y) => x + y, //聚合函数
      windowDuration = Durations.seconds(15), //窗口大小
      slideDuration = Durations.seconds(5) //滑动时间
    )


    val windwsDS1: DStream[(String, Int)] = kvDS.reduceByKeyAndWindow(
      (x, y) => x + y, //聚合函数
      (i, j) => i - j, //减去上一个窗口多余的部分
      windowDuration = Durations.seconds(15), //窗口大小
      slideDuration = Durations.seconds(5) //滑动时间
    )


    //ds api  -->  rdd api

    windwsDS1
      .filter(_._2 != 0)
      .foreachRDD(rdd => {
        rdd.sortBy(-_._2).foreach(println)
      })

    ssc.start()
    ssc.awaitTermination()

  }
}
